/************************************************************************/
// Copyright (C) 2016, han_gangbiao. All rights reserved.
// Use of this source code is governed by a GPL-v2 license that can be found in the License file.
// 
// [Time]:      2016-3-18 21:06:59
// [Author]:    han_gangbiao [http://my.oschina.net/jackhen]
// [Info]:      
/************************************************************************/

#include "rpc_server_base.h"
#include "RpcServerMgr.h"

CRpcServerBase::CRpcServerBase( CRpcServerMgr* pMgr )
    :m_rpc_bean_map(pMgr)
{
    m_pMgr = pMgr;
}

CRpcServerBase::~CRpcServerBase()
{
    stop_rpc_server();
}

int CRpcServerBase::start_rpc_server( const char* local_ip, int local_port )
{
    if(m_pMgr == NULL) return -1;
    m_listenfd = m_pMgr->m_netReactor.open_tcp_server(local_ip, local_port);
    if(m_listenfd < 0)
        return -1;
    m_pMgr->add_rpc_server(this);
    return m_listenfd;
}

int CRpcServerBase::stop_rpc_server()
{
    if(m_pMgr) {
        m_pMgr->m_netReactor.close_fd(m_listenfd);
        m_rpc_bean_map.close_all_rpc_bean();
    }
    return 0;
}

int CRpcServerBase::invoke( SOCK_FD fd, const smart::ref_ptr<NetBinaryPdu>& pdu )
{
    if(pdu == NULL) return -1;
    pdu->toStream();
    if(m_pMgr)
        return m_pMgr->m_netReactor.send(fd, pdu->getBinaryStream(), pdu->getBinaryStreamLength());
    return -1;
}

int CRpcServerBase::on_connect( SOCK_FD cli_fd, const char* remote_ip, int remote_port )
{
    smart::ref_ptr<RpcBean> bean = new RpcBean(cli_fd, remote_ip, remote_port, this);
    m_rpc_bean_map.add_rpc_bean(bean.getPointer());
    return 0;
}
int CRpcServerBase::on_data( SOCK_FD fd, const char* data, int len, const char* remote_ip, int remote_port )
{
    smart::ref_ptr<RpcBean> bean = m_rpc_bean_map.get_rpc_bean(fd);
    if(bean) {
        bean->m_decoder.Parse(fd, data, len);
    } else {
        printf("on_data, but not found in rpc_bean_map\n");
    }
    return 0;
}
int CRpcServerBase::on_send_ack( SOCK_FD fd, int len )
{
    return 0;
}
int CRpcServerBase::on_close( SOCK_FD fd, int error )
{
    printf("CRpcServerBase::on_close: fd=%d, error=%d\n", fd, error);
    m_rpc_bean_map.del_rpc_bean(fd);
    this->on_rpc_close(fd, error);
    return 0;
}
int CRpcServerBase::on_listen_close( SOCK_FD listenfd, int error )
{
    printf("CRpcServerBase::on_listen_close listenfd=[%d],error=[%d]\n", listenfd, error);
    m_rpc_bean_map.del_all_rpc_bean();
    return 0;
}
int CRpcServerBase::onPDU( int fd, smart::ref_ptr<NetBinaryPdu> pdu )
{
    this->on_invoke(fd, pdu);
    return 0;
}
int CRpcServerBase::onError( int fd, BinaryDecoderError_t err )
{
    printf("server decoder fd[%d], error[%d]\n", fd, err);
    return 0;
}


//>RpcBean
RpcBean::RpcBean( SOCK_FD fd, const char* ip, int port, IBinaryDecoderListener* decoder_listener )
    :m_decoder(decoder_listener)
{
    m_fd = fd;
    memset(m_ip, 0, sizeof m_ip);
    strncpy(m_ip, ip, sizeof m_ip);
    m_port = port;
}

RpcBean::~RpcBean() { }


//>RpcBeanMap
RpcBeanMap::RpcBeanMap( CRpcServerMgr* pMgr )
{
    m_pMgr = pMgr;
    m_beans.clear();
}

RpcBeanMap::~RpcBeanMap()
{
    m_beans.clear();
}

smart::ref_ptr<RpcBean> RpcBeanMap::get_rpc_bean(SOCK_FD fd)
{
    RpcBean* bean = NULL;
    smart::guard_t guard(&m_lock);
    std::map< SOCK_FD, smart::ref_ptr<RpcBean> >::iterator it = m_beans.find(fd);
    if(it != m_beans.end())
    {
        bean = (it->second).getPointer();
    }
    return bean;
}
int RpcBeanMap::add_rpc_bean(RpcBean* bean)
{
    if(bean == NULL) return -1;
    smart::guard_t guard(&m_lock);
    std::map< SOCK_FD, smart::ref_ptr<RpcBean> >::iterator it = m_beans.find(bean->m_fd);
    if(it != m_beans.end())
    {
        m_beans.erase(it);
    }
    m_beans[bean->m_fd] = bean;
    return 0;
    
}
int RpcBeanMap::del_rpc_bean(SOCK_FD fd)
{
    smart::guard_t guard(&m_lock);
    std::map< SOCK_FD, smart::ref_ptr<RpcBean> >::iterator it = m_beans.find(fd);
    if(it != m_beans.end())
    {
        m_beans.erase(it);
        return 0;
    }
    return -1;
}

void RpcBeanMap::del_all_rpc_bean()
{
    smart::guard_t guard(&m_lock);
    m_beans.clear();
}

int RpcBeanMap::close_rpc_bean(SOCK_FD fd)
{
    smart::guard_t guard(&m_lock);
    std::map< SOCK_FD, smart::ref_ptr<RpcBean> >::iterator it = m_beans.find(fd);
    if(it != m_beans.end())
    {
        m_pMgr->close_fd(it->second->m_fd);
        return 0;
    }
    return -1;
}

void RpcBeanMap::close_all_rpc_bean()
{
    smart::guard_t guard(&m_lock);
    std::map< SOCK_FD, smart::ref_ptr<RpcBean> >::iterator it = m_beans.begin();
    for(; it!=m_beans.end(); ++it)
    {
        m_pMgr->close_fd(it->second->m_fd);
    }
}
